package com.insight.common.mq.subscribe;

import com.yuanqiao.insight.common.util.common.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import com.insight.common.constant.CacheConstant;
import com.insight.common.mq.RedisMq;
import com.insight.common.mq.RedisMqInit;
import com.insight.common.mq.subscribe.entity.SubscribeMessage;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author JinPengDong
 * @date 2023/12/20
 */
@Slf4j
public abstract class Subscribe {

    private static final ExecutorService pools = Executors.newCachedThreadPool();

    abstract public List<SubscribeMessage> assemblyData();

    public void subscribeMessageEvent(){
        RedisMq redisMq = RedisMqInit.getRedisMq();
        RedisUtils redisUtils = RedisMqInit.getRedisUtils();
        List<SubscribeMessage> messageList = assemblyData();
        final List<CompletableFuture<Integer>> completableFutures = messageList.stream().map(subscribeMessage ->
                CompletableFuture.supplyAsync(() -> {
                    redisMq.subscribe(subscribeMessage.getListener(), subscribeMessage.getStreams());
                    return 1;
                }, pools)).collect(Collectors.toList());

        // 等待所有任务执行完
        CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[messageList.size()])).join();
        AtomicInteger atomicInteger = new AtomicInteger(0);
        for (CompletableFuture<Integer> completableFuture : completableFutures) {
            completableFuture.whenComplete((result, e) -> {
                final int size = atomicInteger.addAndGet(result);
                if (size == messageList.size()) {
                    RedisMq.atomicBoolean.set(true);
                    redisUtils.hset(CacheConstant.SYS_STREAM_HEARTBEAT + "cache", RedisMqInit.getCONSUMER(), RedisMqInit.getGROUP(), 3900);
                    redisUtils.set(CacheConstant.SYS_STREAM_HEARTBEAT + RedisMqInit.getCONSUMER(), RedisMqInit.getGROUP(), 3900);
                    log.info("======已全部订阅RedisMq消息======");
                }
            });
        }
    }
}
